package com.yz.kafka;


import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
@Slf4j
public class ConsumerProducer {

    @Resource
    KafkaTemplate<String, Object> kafkaTemplate;

    /**
     *  创建kafka监听者
     */

    ProducerListener<String, Object> producerListener = new ProducerListener<String, Object>() {

        /**
         * 创建一个生产者对象KafkaProducer
         */
        @Override
        public void onSuccess(ProducerRecord<String, Object> producerRecord, RecordMetadata recordMetadata) {
            log.info("Message send success : " + producerRecord.toString());
        }

    };

    /**
     * 发送信息
     * @param topic   主题
     * @param dataJson  信息
     */
    public void send(String topic, Object dataJson) {

        kafkaTemplate.setProducerListener(producerListener);
        kafkaTemplate.send(topic, dataJson.toString());
    }
}